package org.egomsl.mw.mapred;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapred.RecordWriter;
import org.apache.hadoop.mapred.Reporter;
import org.egomsl.mw.HadoopRecord;
import org.egomsl.mw.region.RegionWriter;

import java.io.IOException;

public class ABaseRecordWriter implements RecordWriter<BytesWritable, HadoopRecord> {

    public RegionWriter regionWriter;

    public ABaseRecordWriter(FileSystem fs, Path outputPath) throws IOException {
        if(!fs.exists(outputPath)) {
            fs.mkdirs(outputPath);
        }

        regionWriter = new RegionWriter(fs, outputPath);
    }

    @Override
    public void write(BytesWritable key, HadoopRecord value) throws IOException {
        if(regionWriter != null) {
            regionWriter.add(value.getRecord());
        }
    }

    @Override
    public void close(Reporter reporter) throws IOException {
        if(regionWriter != null) {
            regionWriter.flush();
        }
    }
}
